#include "gateway/beast_http/http_server.hpp"
#include "gateway/beast_http/router.hpp"
#include "gateway/beast_http/request_adapter.hpp"
#include "gateway/beast_http/response_adapter.hpp"
#include "gateway/beast_http/offload.hpp"
#include "framework/service_bus_v2.h"
#include <boost/asio/signal_set.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <iostream>
#include "logger/LoggerManager.h"
#include <chrono>
#include <boost/beast/websocket.hpp>
#include <jwt.h>
#include <fstream>
#include <sstream>
#include <cstdlib>
#include <memory>
#include <algorithm>
#include <boost/algorithm/string/predicate.hpp>

namespace beast = boost::beast;
namespace http = beast::http;
namespace net = boost::asio;
using tcp = net::ip::tcp;

namespace gateway::beast_http {

HttpServer::HttpServer(const std::string& host,
                       unsigned short port,
                       int io_threads,
                       int worker_threads)
    : host_(host), port_(port), io_threads_(io_threads), worker_threads_(worker_threads) {}

void HttpServer::set_router(std::shared_ptr<Router> router) { router_ = std::move(router); }

void HttpServer::run() {
    if (io_threads_ <= 0) io_threads_ = std::max(1u, std::thread::hardware_concurrency());
    if (worker_threads_ <= 0) worker_threads_ = io_threads_;

    ioc_ = std::make_unique<net::io_context>(io_threads_);
    worker_pool_ = std::make_unique<net::thread_pool>(worker_threads_);

    tcp::endpoint ep{net::ip::make_address(host_), port_};
    acceptor_ = std::make_unique<tcp::acceptor>(*ioc_);
    beast::error_code ec;
    acceptor_->open(ep.protocol(), ec);
    acceptor_->set_option(net::socket_base::reuse_address(true), ec);
    acceptor_->bind(ep, ec);
    acceptor_->listen(net::socket_base::max_listen_connections, ec);

    do_accept();

    // 信号处理器，优雅停止
    net::signal_set signals(*ioc_, SIGINT, SIGTERM);
    signals.async_wait([this](auto, auto){ stop(); });

    std::vector<std::thread> threads;
    threads.reserve(io_threads_);
    for (int i = 0; i < io_threads_; ++i) {
        threads.emplace_back([this]{ ioc_->run(); });
    }
    std::cout << "[HTTP] Listening on " << host_ << ':' << port_ << " (io="
              << io_threads_ << ", workers=" << worker_threads_ << ")" << std::endl;
    for (auto& t : threads) t.join();
}

void HttpServer::stop() {
    beast::error_code ec;
    if (acceptor_) acceptor_->close(ec);
    if (ioc_) ioc_->stop();
    if (worker_pool_) worker_pool_->stop();
}

void HttpServer::do_accept() {
    acceptor_->async_accept(net::make_strand(*ioc_),
        [this](beast::error_code ec, tcp::socket socket){
            if (!ec) start_session(std::move(socket));
            do_accept();
        });
}

static std::string make_uuid() {
    static thread_local boost::uuids::random_generator gen;
    return to_string(gen());
}

bool HttpServer::validate_jwt_token(const std::string& token,
                                    nlohmann::json& claims,
                                    std::string& error) const {
    if (token.empty()) {
        error = "empty token";
        return false;
    }

    auto make_jwt_unique = [](jwt_t* ptr) {
        return std::unique_ptr<jwt_t, decltype(&jwt_free)>{ptr, &jwt_free};
    };

    jwt_t* raw = nullptr;
    if (jwt_decode(&raw, token.c_str(), nullptr, 0) != 0) {
        error = "invalid token format";
        return false;
    }
    auto jwt_holder = make_jwt_unique(raw);

    jwt_alg_t token_alg = jwt_get_alg(jwt_holder.get());
    if (!settings_.jwt.alg.empty()) {
        jwt_alg_t expected_alg = jwt_str_alg(settings_.jwt.alg.c_str());
        if (expected_alg == JWT_ALG_INVAL) {
            error = "unsupported configured algorithm";
            return false;
        }
        if (token_alg != expected_alg) {
            error = "alg mismatch";
            return false;
        }
    }

    std::string kid;
    if (const char* kid_header = jwt_get_header(jwt_holder.get(), "kid")) {
        kid = kid_header;
    }

    if (settings_.jwt.require_signature) {
        std::string key;
        if (token_alg == JWT_ALG_HS256) {
            key = settings_.jwt.secret;
        }
#ifdef OPENSSL_FOUND
        else if (token_alg == JWT_ALG_RS256) {
            auto pem = resolve_rs256_key(kid);
            if (!pem) {
                error = "missing rs256 key";
                return false;
            }
            key = *pem;
        }
#endif
        else {
            error = "unsupported algorithm";
            return false;
        }

        if (key.empty()) {
            error = "signing key not configured";
            return false;
        }

        jwt_holder.reset();
        raw = nullptr;
        if (jwt_decode(&raw,
                       token.c_str(),
                       reinterpret_cast<const unsigned char*>(key.data()),
                       static_cast<int>(key.size())) != 0) {
            error = "invalid token signature";
            return false;
        }
        jwt_holder.reset(raw);
    }

    std::unique_ptr<char, decltype(&std::free)> grants(
        jwt_get_grants_json(jwt_holder.get(), nullptr),
        &std::free);
    if (grants && grants.get()[0] != '\0') {
        try {
            claims = nlohmann::json::parse(grants.get());
        } catch (...) {
            error = "invalid claims json";
            return false;
        }
    } else {
        claims = nlohmann::json::object();
    }

    using std::chrono::system_clock;
    const auto now_sec = std::chrono::duration_cast<std::chrono::seconds>(
                             system_clock::now().time_since_epoch())
                             .count();
    if (claims.contains("exp") && claims["exp"].is_number_integer()) {
        if (now_sec >= claims["exp"].get<long long>()) {
            error = "token expired";
            return false;
        }
    }
    if (claims.contains("nbf") && claims["nbf"].is_number_integer()) {
        if (now_sec < claims["nbf"].get<long long>()) {
            error = "token not active";
            return false;
        }
    }

    if (!settings_.jwt.issuer.empty()) {
        if (!claims.contains("iss") || !claims["iss"].is_string() ||
            claims["iss"].get<std::string>() != settings_.jwt.issuer) {
            error = "invalid issuer";
            return false;
        }
    }
    if (!settings_.jwt.audience.empty()) {
        bool audience_ok = false;
        if (claims.contains("aud")) {
            if (claims["aud"].is_string()) {
                const auto aud = claims["aud"].get<std::string>();
                audience_ok = std::find(settings_.jwt.audience.begin(),
                                        settings_.jwt.audience.end(),
                                        aud) != settings_.jwt.audience.end();
            } else if (claims["aud"].is_array()) {
                for (const auto& entry : claims["aud"]) {
                    if (entry.is_string()) {
                        const auto aud = entry.get<std::string>();
                        if (std::find(settings_.jwt.audience.begin(),
                                      settings_.jwt.audience.end(),
                                      aud) != settings_.jwt.audience.end()) {
                            audience_ok = true;
                            break;
                        }
                    }
                }
            }
        }
        if (!audience_ok) {
            error = "invalid audience";
            return false;
        }
    }

    return true;
}

#ifdef OPENSSL_FOUND
std::optional<std::string> HttpServer::load_pem_from_file(const std::string& path) const {
    if (path.empty()) return std::nullopt;
    std::lock_guard<std::mutex> lk(rs256_key_mutex_);
    auto it = rs256_key_cache_.find(path);
    if (it != rs256_key_cache_.end()) return it->second;

    std::ifstream ifs(path);
    if (!ifs.is_open()) {
        return std::nullopt;
    }
    std::ostringstream oss;
    oss << ifs.rdbuf();
    auto pem = oss.str();
    if (pem.empty()) {
        return std::nullopt;
    }
    auto inserted = rs256_key_cache_.emplace(path, pem);
    return inserted.first->second;
}

std::optional<std::string> HttpServer::resolve_rs256_key(const std::string& kid) const {
    if (!kid.empty()) {
        auto it = settings_.jwt.key_paths.find(kid);
        if (it != settings_.jwt.key_paths.end()) {
            if (auto pem = load_pem_from_file(it->second)) {
                return pem;
            }
        }
    }
    return load_pem_from_file(settings_.jwt.public_key_path);
}
#endif

void HttpServer::log_access(const std::string& remote,
                            const std::string& req_line,
                            unsigned status,
                            std::size_t bytes,
                            long long elapsed_ms,
                            const std::string& request_id) const {
    if (!settings_.access_log_enable) return;
    // 结构化日志（可替换为 JSON 序列化）
    LOG_INFO << "{\"remote\":\"" << remote
             << "\",\"req\":\"" << req_line
             << "\",\"status\":" << status
             << ",\"bytes\":" << bytes
             << ",\"elapsed_ms\":" << elapsed_ms
             << ",\"rid\":\"" << request_id << "\"}";
}

bool HttpServer::allow_request() {
    if (!settings_.rl.enable || settings_.rl.rps <= 0) return true;
    using std::chrono::system_clock;
    auto now = system_clock::now();
    auto sec = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
    auto cur = static_cast<std::uint64_t>(sec);
    auto prev = rl_sec_.load(std::memory_order_relaxed);
    if (cur != prev) {
        rl_sec_.store(cur, std::memory_order_relaxed);
        rl_count_.store(0, std::memory_order_relaxed);
    }
    auto c = rl_count_.fetch_add(1, std::memory_order_relaxed) + 1;
    return c <= static_cast<std::uint64_t>(settings_.rl.rps);
}

bool HttpServer::allow_request_key(const std::string& key) {
    if (!settings_.rl.enable || settings_.rl.rps <= 0) return true;
    using std::chrono::system_clock;
    auto now = system_clock::now();
    auto sec = static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count());
    std::lock_guard<std::mutex> lk(rl_mutex_);
    auto& b = rl_buckets_[key];
    // 令牌桶：每秒补充 rps 个，容量 rps + burst
    if (b.sec != sec) {
        int capacity = settings_.rl.rps + std::max(0, settings_.rl.burst);
        int add = settings_.rl.rps * static_cast<int>(b.sec == 0 ? 1 : (sec - b.sec));
        b.tokens = std::min(capacity, b.tokens + (b.sec == 0 ? capacity : add));
        b.sec = sec;
    }
    if (b.tokens <= 0) return false;
    --b.tokens;
    return true;
}

// 授权：基于路径前缀与角色
bool HttpServer::authorize_path(const std::string& path, const nlohmann::json& claims) const {
    if (settings_.authz.rules.empty()) return true;
    std::vector<std::string> have_roles;
    try {
        if (claims.contains("roles") && claims["roles"].is_array()) {
            for (const auto& r : claims["roles"]) have_roles.push_back(r.get<std::string>());
        } else if (claims.contains("role")) {
            if (claims["role"].is_string()) have_roles.push_back(claims["role"].get<std::string>());
            else if (claims["role"].is_array()) for (const auto& r : claims["role"]) have_roles.push_back(r.get<std::string>());
        }
    } catch (...) {}
    for (const auto& rule : settings_.authz.rules) {
        if (path.rfind(rule.prefix, 0) == 0) {
            if (rule.roles.empty()) return true;
            for (const auto& need : rule.roles) {
                for (const auto& have : have_roles) if (have == need) return true;
            }
            return false;
        }
    }
    return true;
}

http::response<http::string_body> HttpServer::make_metrics_response(unsigned http_version) const {
    http::response<http::string_body> res;
    res.version(http_version);
    res.set(http::field::content_type, "text/plain; version=0.0.4");
    std::ostringstream oss;
    oss << "# HELP gateway_http_requests_total Total HTTP requests\n";
    oss << "# TYPE gateway_http_requests_total counter\n";
    oss << "gateway_http_requests_total " << metric_req_total_.load() << "\n";
    oss << "# HELP gateway_http_responses_total Total HTTP responses by class\n";
    oss << "# TYPE gateway_http_responses_total counter\n";
    oss << "gateway_http_responses_total{class=\"2xx\"} " << metric_resp_2xx_.load() << "\n";
    oss << "gateway_http_responses_total{class=\"4xx\"} " << metric_resp_4xx_.load() << "\n";
    oss << "gateway_http_responses_total{class=\"5xx\"} " << metric_resp_5xx_.load() << "\n";
    oss << "# HELP gateway_http_bytes_out_total Total bytes sent\n";
    oss << "# TYPE gateway_http_bytes_out_total counter\n";
    oss << "gateway_http_bytes_out_total " << metric_bytes_out_.load() << "\n";
    oss << "# HELP gateway_http_in_flight In-flight requests\n";
    oss << "# TYPE gateway_http_in_flight gauge\n";
    oss << "gateway_http_in_flight " << metric_in_flight_.load() << "\n";
    oss << "# HELP gateway_ws_messages_total Total WebSocket messages\n";
    oss << "# TYPE gateway_ws_messages_total counter\n";
    oss << "gateway_ws_messages_total " << metric_ws_msg_total_.load() << "\n";
    // Histogram
    oss << "# HELP gateway_http_request_duration_ms Request latency\n";
    oss << "# TYPE gateway_http_request_duration_ms histogram\n";
    uint64_t cumulative = 0;
    for (size_t i = 0; i < kLatencyBounds.size(); ++i) {
        cumulative += metric_latency_buckets_[i].load();
        oss << "gateway_http_request_duration_ms_bucket{le=\"" << kLatencyBounds[i] << "\"} " << cumulative << "\n";
    }
    oss << "gateway_http_request_duration_ms_bucket{le=\"+Inf\"} " << metric_latency_count_.load() << "\n";
    oss << "gateway_http_request_duration_ms_sum " << metric_latency_sum_ms_.load() << "\n";
    oss << "gateway_http_request_duration_ms_count " << metric_latency_count_.load() << "\n";
    // per-service totals
    {
        std::lock_guard<std::mutex> lk(metric_mtx_);
        for (const auto& kv : metric_service_total_) {
            oss << "gateway_http_requests_total{service=\"" << kv.first << "\"} " << kv.second << "\n";
        }
    }
    res.body() = oss.str();
    res.prepare_payload();
    return res;
}

void HttpServer::start_session(tcp::socket socket) {
    net::co_spawn(*ioc_,
        [this, s = std::move(socket)]() mutable -> net::awaitable<void> {
            beast::flat_buffer buffer;
            for (;;) {
                http::request<http::string_body> req;
                beast::error_code ec;
                co_await http::async_read(s, buffer, req, net::redirect_error(net::use_awaitable, ec));
                if (ec == http::error::end_of_stream || ec == net::error::operation_aborted) break;
                if (ec) break;

                auto t0 = std::chrono::steady_clock::now();
                metric_req_total_.fetch_add(1, std::memory_order_relaxed);
                metric_in_flight_.fetch_add(1, std::memory_order_relaxed);
                auto path = std::string(req.target());
                auto svc = router_ ? router_->match(path) : std::optional<std::string>{};
                http::response<http::string_body> res;
                std::string rid = make_uuid();
                std::string remote_ip = s.remote_endpoint().address().to_string();
                std::string req_line = req.method_string().to_string() + " " + path;
                // helper in advance to avoid goto crossing init
                auto prefix_match = [](const std::vector<std::string>& arr, const std::string& p){ for (const auto& x:arr) if (p.rfind(x,0)==0) return true; return false; };
                if (path == "/metrics") {
                    res = make_metrics_response(req.version());
                    goto WRITE_RESP;
                }
                // 白名单/黑名单（IP/路径）
                if (!settings_.rl.whitelist_ips.empty()) {
                    bool ok = std::find(settings_.rl.whitelist_ips.begin(), settings_.rl.whitelist_ips.end(), remote_ip) != settings_.rl.whitelist_ips.end();
                    if (!ok) {
                        res.version(req.version()); res.result(http::status::forbidden);
                        res.set(http::field::content_type, "application/json"); res.set("X-Request-Id", rid);
                        res.body() = "{\"error\":\"ip not allowed\"}"; res.prepare_payload();
                        co_await http::async_write(s, res, net::use_awaitable);
                        auto t1 = std::chrono::steady_clock::now();
                        auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
                        metric_resp_4xx_.fetch_add(1, std::memory_order_relaxed);
                        metric_bytes_out_.fetch_add(res.body().size(), std::memory_order_relaxed);
                        log_access(remote_ip, req_line, static_cast<unsigned>(res.result_int()), res.body().size(), ms, rid);
                        metric_in_flight_.fetch_sub(1, std::memory_order_relaxed);
                        continue;
                    }
                }
                if (std::find(settings_.rl.blacklist_ips.begin(), settings_.rl.blacklist_ips.end(), remote_ip)!=settings_.rl.blacklist_ips.end() ||
                    prefix_match(settings_.rl.blacklist_paths, path)) {
                    res.version(req.version()); res.result(http::status::forbidden);
                    res.set(http::field::content_type, "application/json"); res.set("X-Request-Id", rid);
                    res.body() = "{\"error\":\"forbidden\"}"; res.prepare_payload();
                    co_await http::async_write(s, res, net::use_awaitable);
                    auto t1 = std::chrono::steady_clock::now();
                    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
                    metric_resp_4xx_.fetch_add(1, std::memory_order_relaxed);
                    metric_bytes_out_.fetch_add(res.body().size(), std::memory_order_relaxed);
                    log_access(remote_ip, req_line, static_cast<unsigned>(res.result_int()), res.body().size(), ms, rid);
                    metric_in_flight_.fetch_sub(1, std::memory_order_relaxed);
                    continue;
                }
                if (!settings_.rl.whitelist_paths.empty() && !prefix_match(settings_.rl.whitelist_paths, path)) {
                    res.version(req.version()); res.result(http::status::forbidden);
                    res.set(http::field::content_type, "application/json"); res.set("X-Request-Id", rid);
                    res.body() = "{\"error\":\"path not allowed\"}"; res.prepare_payload();
                    co_await http::async_write(s, res, net::use_awaitable);
                    auto t1 = std::chrono::steady_clock::now();
                    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
                    metric_resp_4xx_.fetch_add(1, std::memory_order_relaxed);
                    metric_bytes_out_.fetch_add(res.body().size(), std::memory_order_relaxed);
                    log_access(remote_ip, req_line, static_cast<unsigned>(res.result_int()), res.body().size(), ms, rid);
                    metric_in_flight_.fetch_sub(1, std::memory_order_relaxed);
                    continue;
                }

                // rate limit
                // 限流：支持 global | ip | path | ip_path
                if (settings_.rl.enable) {
                    std::string key;
                    if (settings_.rl.mode == "global") key = "global";
                    else if (settings_.rl.mode == "ip") key = remote_ip;
                    else if (settings_.rl.mode == "path") key = path;
                    else key = remote_ip + "|" + path;
                    if (!allow_request_key(key)) {
                        res.version(req.version());
                        res.result(http::status::too_many_requests);
                        res.set(http::field::content_type, "application/json");
                        res.set("X-Request-Id", rid);
                        res.body() = "{\"error\":\"rate limit exceeded\"}";
                        res.prepare_payload();
                        co_await http::async_write(s, res, net::use_awaitable);
                        auto t1 = std::chrono::steady_clock::now();
                        auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
                        auto status_code = static_cast<unsigned>(res.result_int());
                        metric_resp_4xx_.fetch_add(1, std::memory_order_relaxed);
                        metric_bytes_out_.fetch_add(res.body().size(), std::memory_order_relaxed);
                        log_access(remote_ip, req_line, status_code, res.body().size(), ms, rid);
                        metric_in_flight_.fetch_sub(1, std::memory_order_relaxed);
                        continue;
                    }
                }
                if (!svc) {
                    res.version(req.version());
                    res.result(http::status::not_found);
                    res.set(http::field::content_type, "application/json");
                    res.set("X-Request-Id", rid);
                    res.body() = "{\"error\":\"not found\"}";
                    res.prepare_payload();
                } else if (beast::websocket::is_upgrade(req)) {
                    // WebSocket 升级
                    beast::websocket::stream<tcp::socket> ws(std::move(s));
                    co_await ws.async_accept(req, net::use_awaitable);
                    for (;;) {
                        beast::flat_buffer wsbuf;
                        boost::system::error_code wec;
                        co_await ws.async_read(wsbuf, net::redirect_error(net::use_awaitable, wec));
                        if (wec) break; // 连接关闭或错误

                        // 将文本消息解析为 JSON，请求到后端
                        std::string text = beast::buffers_to_string(wsbuf.data());
                        std::string reply_json;
                        try {
                            auto j = nlohmann::json::parse(text);
                            std::string path = j.value("path", "/api/ping");
                            std::string method = j.value("method", "GET");
                            cpp_backend::framework::Request fw_req;
                            fw_req.SetPath(path);
                            fw_req.SetMethod(method);
                            if (j.contains("headers") && j["headers"].is_object()) {
                                for (auto it = j["headers"].begin(); it != j["headers"].end(); ++it) {
                                    fw_req.SetHeader(it.key(), it.value().dump());
                                }
                            }
                            if (j.contains("params") && j["params"].is_object()) {
                                for (auto it = j["params"].begin(); it != j["params"].end(); ++it) {
                                    fw_req.SetParam(it.key(), it.value().dump());
                                }
                            }
                            if (j.contains("body")) {
                                fw_req.SetBody(j["body"]);
                            }
                            fw_req.SetClientIp("ws");
                            fw_req.SetRequestId(make_uuid());

                            // 路由：复用匹配结果 svc
                            auto fw_resp = co_await co_offload<cpp_backend::framework::Response>(
                                *ioc_, *worker_pool_, [svc_name=*svc, fw_req]() mutable {
                                    auto& bus = cpp_backend::framework::ServiceBusV2::Instance();
                                    return bus.CallService(svc_name, fw_req);
                                });
                            reply_json = fw_resp.ToString();
                        } catch (const std::exception& e) {
                            reply_json = std::string("{\"error\":\"") + e.what() + "\"}";
                        }
                        ws.text(true);
                        co_await ws.async_write(net::buffer(reply_json), net::use_awaitable);
                        metric_ws_msg_total_.fetch_add(1, std::memory_order_relaxed);
                    }
                    // WS 模式完成后退出会话
                    co_return;
                } else {
                    // 适配并投递到线程池执行同步 gRPC 调用
                    try {
                        auto fw_req = RequestAdapter::ToFramework(req, remote_ip, rid);
                        // JWT（可选）：从 Authorization: Bearer <token> 解析基本声明
                        if (settings_.jwt.enable) {
                            auto it = req.find(http::field::authorization);
                            if (it != req.end()) {
                                std::string auth = std::string(it->value());
                                const std::string bearer = "Bearer ";
                                if (boost::istarts_with(auth, bearer)) {
                                    std::string token = auth.substr(bearer.size());
                                    nlohmann::json claims;
                                    std::string jwt_error;
                                    if (!validate_jwt_token(token, claims, jwt_error)) {
                                        res.version(req.version());
                                        res.result(http::status::unauthorized);
                                        res.set(http::field::content_type, "application/json");
                                        res.set("X-Request-Id", rid);
                                        res.body() = nlohmann::json{{"error", jwt_error}}.dump();
                                        res.prepare_payload();
                                        goto WRITE_RESP;
                                    }

                                    if (claims.contains("uid") && claims["uid"].is_number_integer()) {
                                        fw_req.SetUserId(claims["uid"].get<long long>());
                                    } else if (claims.contains("sub")) {
                                        if (claims["sub"].is_number_integer()) {
                                            fw_req.SetUserId(claims["sub"].get<long long>());
                                        } else if (claims["sub"].is_string()) {
                                            try {
                                                fw_req.SetUserId(std::stoll(claims["sub"].get<std::string>()));
                                            } catch (...) {
                                            }
                                        }
                                    }
                                    if (claims.contains("name")) {
                                        if (claims["name"].is_string()) {
                                            fw_req.SetUserName(claims["name"].get<std::string>());
                                        } else {
                                            fw_req.SetUserName(claims["name"].dump());
                                        }
                                    }
                                    if (claims.contains("role")) {
                                        if (claims["role"].is_string()) {
                                            fw_req.SetUserRole(claims["role"].get<std::string>());
                                        } else {
                                            fw_req.SetUserRole(claims["role"].dump());
                                        }
                                    }

                                    if (!authorize_path(path, claims)) {
                                        res.version(req.version());
                                        res.result(http::status::forbidden);
                                        res.set(http::field::content_type, "application/json");
                                        res.set("X-Request-Id", rid);
                                        res.body() = "{\"error\":\"forbidden\"}";
                                        res.prepare_payload();
                                        goto WRITE_RESP;
                                    }
                                }
                            }
                        }
                        // 简化版：future + get；在 io_threads>=2 时仍可保证 I/O 不被整体卡住
                        auto fut = post_to_pool(*worker_pool_, [svc_name=*svc, fw_req]() mutable {
                            auto& bus = cpp_backend::framework::ServiceBusV2::Instance();
                            return bus.CallService(svc_name, fw_req);
                        });
                        auto fw_resp = fut.get();
                        {
                            // per-service metrics
                            if (svc) { std::lock_guard<std::mutex> lk(metric_mtx_); metric_service_total_[*svc]++; }
                            res = ResponseAdapter::ToBeast(fw_resp);
                        }
                        res.version(req.version());
                        res.set("X-Request-Id", rid);
                    } catch (const std::exception& e) {
                        res.version(req.version());
                        res.result(http::status::internal_server_error);
                        res.set(http::field::content_type, "application/json");
                        res.set("X-Request-Id", rid);
                        res.body() = std::string("{\"error\":\"") + e.what() + "\"}";
                        res.prepare_payload();
                    }
                }

WRITE_RESP:
                co_await http::async_write(s, res, net::use_awaitable);
                if (res.need_eof()) break;
                auto t1 = std::chrono::steady_clock::now();
                auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t0).count();
                auto status_code = static_cast<unsigned>(res.result_int());
                if (status_code >= 200 && status_code < 300) metric_resp_2xx_.fetch_add(1, std::memory_order_relaxed);
                else if (status_code >= 400 && status_code < 500) metric_resp_4xx_.fetch_add(1, std::memory_order_relaxed);
                else if (status_code >= 500) metric_resp_5xx_.fetch_add(1, std::memory_order_relaxed);
                metric_bytes_out_.fetch_add(res.body().size(), std::memory_order_relaxed);
                // 直方图统计
                metric_latency_sum_ms_.fetch_add(ms, std::memory_order_relaxed);
                metric_latency_count_.fetch_add(1, std::memory_order_relaxed);
                for (size_t i = 0; i < kLatencyBounds.size(); ++i) {
                    if (ms <= kLatencyBounds[i]) { metric_latency_buckets_[i].fetch_add(1, std::memory_order_relaxed); break; }
                }
                log_access(remote_ip, req_line, status_code, res.body().size(), ms, rid);
                metric_in_flight_.fetch_sub(1, std::memory_order_relaxed);
            }
            beast::error_code ec;
            s.shutdown(tcp::socket::shutdown_send, ec);
            co_return;
        },
        net::detached);
}

} // namespace gateway::beast_http
